package com.qyer.commons.concurrent.assemblingline;

import static com.qyer.commons.concurrent.assemblingline.AssemblingLineStatus.INITED;
import static com.qyer.commons.concurrent.assemblingline.AssemblingLineStatus.STANDBY;
import static com.qyer.commons.concurrent.assemblingline.ProcessMode.SINGLE;
import static com.qyer.commons.utils.CommonUtils.timeCost;

import com.qyer.commons.concurrent.ArrayListBulk;
import com.qyer.commons.concurrent.GenericLinkedBlockingQueue;
import com.qyer.commons.mc.converter.Converter;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;

/**
 * User: Z J Wu Date: 2016/06/02 Time: 17:19 Package: com.qyer.commons.concurrent
 */
public class FileBulkRawDataGenerator<T> extends RawDataGenerator<T> {

  private static final Logger LOGGER = LoggerFactory.getLogger(FileBulkRawDataGenerator.class);
  private String inputPath;

  private Converter<String, T> converter;

  private BufferedReader br;

  public FileBulkRawDataGenerator(String name, StatisticContext context, String inputPath,
                                  Converter<String, T> converter) {
    super(name, context);
    this.inputPath = inputPath;
    this.converter = converter;
  }

  @Override
  public void close() {
    if (br != null) {
      try {
        br.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }

  @Override
  protected void beginProduce(GenericLinkedBlockingQueue<ArrayListBulk<T>> mainQueue) throws
    Exception {
    br = new BufferedReader(new InputStreamReader(new FileInputStream(new File(inputPath))));
    String line;
    ArrayListBulk<T> bulk = ArrayListBulk.newArrayListBulkByCapacity(bulkSize);
    while ((line = br.readLine()) != null) {
      long t1 = System.currentTimeMillis();
      if (StringUtils.isBlank(line)) {
        continue;
      }
      T t;
      try {
        t = converter.convert(line);
        context.reportSuccess(name, SINGLE);
      } catch (Exception e) {
        LOGGER.warn("Cannot convert current line({}).", line);
        context.reportFailure(name, SINGLE);
        continue;
      } finally {
        context.reportTimeCost(name, timeCost(t1), SINGLE);
      }
      if (bulk.isFull()) {
        mainQueue.put(bulk);
        bulk = ArrayListBulk.newArrayListBulkByCapacity(bulkSize);
      }
      bulk.add(t);
    }
    // 最后要清空
    mainQueue.put(bulk);
  }

  @Override
  public AssemblingLineStatus check() {
    if (INITED.equals(super.check())) {
      return INITED;
    }
    if (bulkSize <= 0 || StringUtils.isEmpty(inputPath) || converter == null) {
      return INITED;
    }
    return STANDBY;
  }

}